package com.example.mq.mqsever.datacenter;

/**
 * Created with IntelliJ IDEA.
 * Description:
 * User: WHY
 * Date: 2023-09-03
 * Time: 20:37
 */

import com.example.mq.common.MqException;
import com.example.mq.mqsever.core.Binding;
import com.example.mq.mqsever.core.Exchange;
import com.example.mq.mqsever.core.MSGQueue;
import com.example.mq.mqsever.core.Message;
import org.springframework.web.bind.annotation.ExceptionHandler;
import sun.awt.image.ImageWatched;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 使用这个类来统一管理内存中的所有数据
 */
public class MemoryDataCenter {
    //key是exchangeName,,value是exchange对象
    //由于这个地方可能会有多个请求,比如增删改查啥的,就涉及到多线程,因此我们用线程安全的容器,ConcurrentHashMap
    private ConcurrentHashMap<String, Exchange> exchangeMap = new ConcurrentHashMap<>();

    //key是queueName,value是MSGQueue对象
    private ConcurrentHashMap<String, MSGQueue> queueMap = new ConcurrentHashMap<>();

    //第一个key是exchangeName,第二个key是queueName,value是binding对象
    private ConcurrentHashMap<String, ConcurrentHashMap<String, Binding>> bindingsMap = new ConcurrentHashMap<>();
    //key是MessageId,value是Message对象
    private ConcurrentHashMap<String, Message> messageMap = new ConcurrentHashMap<>();
    //key是queueName,value是一个Message的链表
    private ConcurrentHashMap<String, LinkedList<Message>> queueMessageMap = new ConcurrentHashMap<>();
    //待确认的消息
//第一个key是queueName,第二个Key是messageId
    private ConcurrentHashMap<String, ConcurrentHashMap<String, Message>> queueMessageWaitAckMap = new ConcurrentHashMap<>();


    public void insertExchange(Exchange exchange) {
        exchangeMap.put(exchange.getName(), exchange);
        System.out.println("[MessageDataCenter] 添加交换机成功 exchange=" + exchange.getName());
    }

    public Exchange getExchange(String exchangeName) {
        return exchangeMap.get(exchangeName);
    }

    public void deleteExchange(String exchangeName) {
        exchangeMap.remove(exchangeName);
        System.out.println("[MessageDataCenter] 删除交换机成功! exchange=" + exchangeName);
    }

    public void insertQueue(MSGQueue queue) {
        queueMap.put(queue.getName(), queue);
        System.out.println("[MessageDataCenter] 添加队列成功! queue=" + queue.getName());
    }

    public MSGQueue getQueue(String queueName) {
        return queueMap.get(queueName);
    }

    public void deleteQueue(String queueName) {
        queueMap.remove(queueName);
        System.out.println("[MessageDataCenter] 删除队列成功!=" + queueName);
    }

    public void insertBinding(Binding binding) throws MqException {
        //先使用exchangeName查一下,对应的哈希表是否存在,不存在就创建一个

//        ConcurrentHashMap<String,Binding> bindingMap=new ConcurrentHashMap<>();
//        if(bindingMap==null){
//            bindingMap=new ConcurrentHashMap<>();
//            bindingsMap.put(binding.getExchangeName(),bindingMap);
//        }
        //这一段等价于上述的方法
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.computeIfAbsent(binding.getExchangeName(), k -> new ConcurrentHashMap<>());
        //再根据queueName查询一下是否存在,如果存在就抛出异常,不存在才能插入
        synchronized (bindingMap) {//不针对bindingsMap加锁,主要是为了减小锁冲突,降低锁的粒度
            if (bindingMap.get(binding.getQueueName()) != null) {
                throw new MqException("[MemoryDataCenter]绑定已经存在! exchangeName=" + binding.getExchangeName()
                        + ",queueName=" + binding.getQueueName());
            }
            bindingMap.put(binding.getQueueName(), binding);

        }
        System.out.println("[MemoryDataCenter] 新绑定添加成功! exchangeName" + binding.getExchangeName() +
                ", queueName=" + binding.getQueueName());
    }

    //获取绑定,写两个版本
    //1.根据exchangeName和queueName确定唯一一个Binding

    public Binding getBinding(String exchangeName, String queueName) {
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(exchangeName);
        if (bindingMap == null) {
            return null;
        }
        return bindingMap.get(queueName);

    }
    //2.根据exchangeName获取到所有的Binding
    public ConcurrentHashMap<String, Binding> getBindings(String exchangeName) {
        return bindingsMap.get(exchangeName);
    }

    public void deleteBinding(Binding binding) throws MqException {
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(binding.getExchangeName());
        if (bindingMap == null) {
            //该交换机没绑定任何队列,报错
            throw new MqException("[MemoryDataCenter] 绑定不存在! exchangeName " + binding.getExchangeName()
                    + ", queueName" + binding.getQueueName());
        }
        bindingMap.remove(binding.getQueueName());

        System.out.println("[MemoryDataCenter] 绑定删除成功! exchangeName" + binding.getExchangeName() +
                ", queueName=" + binding.getQueueName());

    }

    //添加消息
    public void addMessage(Message message) {
        messageMap.put(message.getMessageId(), message);
        System.out.println("[MemoryDataCenter] 新消息添加成功! messageId=" + message.getMessageId());
    }

    //根据id查询消息
    public Message getMessage(String messageId) {
        return messageMap.get(messageId);
    }

    //根据id删除消息
    public void removeMessage(String messageId) {
        messageMap.remove(messageId);
        System.out.println("[MessageDataCenter] 消息被移除! messageId=" + messageId);
    }

    //发送消息到指定队列
    public void sendMessage(MSGQueue queue, Message message) {
        //把消息放到对应的队列数据结构中
        //先根据队列的名字,找到该队列对应的消息链表,没有的话就创建
        //.computeIfAbsent是线程安全的
        LinkedList<Message> messages = queueMessageMap.computeIfAbsent(queue.getName(), k -> new LinkedList<>());
        synchronized (messages) {
            //再把数据添加到messages里面
            messages.add(message);

        }
        //把上述消息插入到总的消息中
        //如果相同的messageId对应的messageId内容一定是一样的
        //服务器代码不会对Message内容(basicProperties和body [])做修改
        addMessage(message);
        System.out.println("[MessageDataCenter] 消息被投递到队列中 messageId" + message.getMessageId());
    }

    //从队列中取消息
    public Message pollMessage(String queueName) {
        //根据队列名,查找一下,对应得队列的消息链表
        LinkedList<Message> messages = queueMessageMap.get(queueName);
        if (messages == null) {
            return null;
        }
        synchronized (messages) {
            //如果消息长度也为空,返回空
            if (messages.size() == 0) {
                return null;
            }
            //链表中有元素
            Message currentMessage=   messages.remove(0);
            System.out.println("[MemoryDataCenter] 消息从队列中取出 messageId="+currentMessage.getMessageId());
                return currentMessage;
        }
    }
    //获取指定队列中消息的个数
    public int getMessageCount(String queueName){
        LinkedList<Message> messages=queueMessageMap.get(queueName);
        if(messages==null){
            //队列中没有消息
            return 0;
        }
        synchronized (messages){
            return messages.size();
        }
    }
    //添加未确认的消息
    public void addMessageWaitACK(String queueName,Message message){
        ConcurrentHashMap<String ,Message> messageHashMap=queueMessageWaitAckMap.computeIfAbsent(queueName,
                k->new ConcurrentHashMap<>());
        messageHashMap.put(message.getMessageId(),message);
        System.out.println("[MessageDataCenter] 消息进入待确认队列! messageId="+message.getMessageId());

    }



    //删除未确认的消息()

    public void removeMessageWaitACK(String queueName, String messageId){
        ConcurrentHashMap<String,Message> messageHashMap=queueMessageWaitAckMap.get(queueName);
        if(messageHashMap==null){
            return;
        }
        messageHashMap.remove(messageId);

        System.out.println("[MemoryDataCenter] 消息从待确认队列删除! messageId="+messageId);
    }
    //获取指定的未确认的消息
    public Message getMessageWaitAck(String queueName,String messageId){
      ConcurrentHashMap<String,Message> messageHashMap=queueMessageWaitAckMap.get(queueName);
      if(messageHashMap==null){
          return null;
      }
      return messageHashMap.get(messageId);
    }

    //这个方法就是从硬盘上读取数据,把硬盘之中持有化在各个维度是的数据都恢复到内存中
    public void recovery(DiskDataCenter diskDataCenter) throws IOException, MqException, ClassNotFoundException {
        //0.清空之前的所有数据
        exchangeMap.clear();
        queueMap.clear();
        bindingsMap.clear();
        messageMap.clear();
        queueMessageMap.clear();

        //1.恢复所有的交换机数据
        List<Exchange> exchanges=diskDataCenter.selectAllExchanges();
        for(Exchange exchange:exchanges){
            exchangeMap.put(exchange.getName(),exchange);

        }
        //2.恢复所有的队列数据
        List<MSGQueue> queues=diskDataCenter.selectAllQueues();
        for(MSGQueue queue:queues){
            queueMap.put(queue.getName(),queue);
        }
        //3.恢复所有的绑定数据
        List<Binding> bindings=diskDataCenter.selectAllBindings();
        for(Binding binding:bindings){
            //根据exchangeName找到哈希表,再插入
            ConcurrentHashMap<String,Binding>  bindingMap=bindingsMap.computeIfAbsent(binding.getExchangeName(),k->
                    new ConcurrentHashMap<>());

            bindingMap.put(binding.getQueueName(),binding);
        }
        //4.恢复所有的消息数据
        //消息在队列里面,所以要遍历所有的队列,获取到所有的消息

        for(MSGQueue queue:queues){
        LinkedList<Message> messages=diskDataCenter.loadAllMessageFromQueue(queue.getName());
        queueMessageMap.put(queue.getName(),messages);
        for(Message message:messages){
            messageMap.put(message.getMessageId(),message);
        }
        }
        //针对未确认的消息这部分内存的数据,不需要从硬盘恢复,之前考虑硬盘存储,也没设计这一块
        //一旦在等待ack的过程中,服务器重启了,此时这些"未被确认的消息",就恢复成"未被取走的消息"
        //因为在硬盘上存储的时候,就是当做"未被取走".不存在什么未被确认这一说



    }




}




